package com.jw.debezuim.websocket;

import com.jw.debezuim.config.filter.CommonFilter;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.PongMessage;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 单实例
 *
 * @Description: websocket处理器
 */
@Component()
@Slf4j
@ConditionalOnProperty(value = {"biz.server.web-socket.enable"})
public class SimulateWebSocketHandler extends TextWebSocketHandler implements ApplicationContextAware {

    /**
     * 在线用户列表<userId,session>
     */
    private final Map<Long, WebSocketSession> users = new ConcurrentHashMap<>();

    public static volatile Map<WebSocketSession, Object> handBizSessionMap = new ConcurrentHashMap<>();
    public static final AtomicInteger connNum = new AtomicInteger();

    /**
     * 执行器线程池
     */
    @Autowired
    private Executor executor;

    /**
     * spring上下文
     */
    private ApplicationContext applicationContext;

    @Autowired
    private ApplicationEventPublisher applicationEventPublisher;
    /**
     * @param
     * @return
     * @description[新增socket,将session添加到用户列表(users),并触发监听器init方法]
     * @author jw
     * @date 2018/8/6 11:26
     **/
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        try {
//            LoginUserDto loginUserDto = this.getLoginUser(session);
            int i = connNum.incrementAndGet();
            log.info("建立连接,当前在线人数：[{}]", i);
        } catch (Exception e) {
            session.close();
            log.error("建立连接异常捕获", e);
        }
    }

    @Override
    protected void handlePongMessage(WebSocketSession session, PongMessage message) throws Exception {
        if (session.isOpen()) {
            session.sendMessage(new PongMessage());
        }
    }

    /**
     * @description[接收socket信息]
     **/
    @Override
    public void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        try {
            String requestId = UUID.randomUUID().toString();
            MDC.put(CommonFilter.REQUESTID, requestId);//日志格式添加请求链
            synchronized (session) {
                if (!session.isOpen()) {
                    return;
                }
            }
        } catch (Exception e) {
            log.error("处理消息异常捕获", e);
            session.close(CloseStatus.SERVER_ERROR);
        } finally {
            MDC.clear();
        }
    }

    @Override
    public void handleTransportError(WebSocketSession session, Throwable e) throws Exception {
        if (session.isOpen()) {
            session.close();
        }
        log.error("连接异常捕获", e);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        int i = connNum.decrementAndGet();
        this.handBizSessionMap.remove(session);
        synchronized (session) {
            session.notifyAll();
        }
        log.info("连接断开,当前连接数:[{}]", i);
        log.info("连接断开,当前处理业务数：[{}]", this.handBizSessionMap.size());
    }

    @Override
    public boolean supportsPartialMessages() {
        return false;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

}
